using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading.Tasks;
using MessageQueues.Abstractions;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
using ObjectSerializers.Abstractions;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace MessageQueues.RabbitMQ
{
    public class Context : BasicMqContext
    {
        private static ConnectionFactory _factory;

        private readonly IConfiguration _configuration;
        private readonly ISerializerContext _serializerContext;
        private readonly IConnection _connection;
        private readonly IDictionary<string, IModel> _channelMap;
        private readonly IDictionary<string, IBasicProperties> _propertiesMap;

        public Context(ILogger<Context> logger, IConfiguration configuration, ISerializerContext serializerContext)
            : base(logger)
        {
            _configuration = configuration;
            _serializerContext = serializerContext;

            _factory ??= this.GetConnectionFactory();
            _connection = _factory.CreateConnection(); //创建账号后记得给权限，否则连接不起
            _channelMap = new ConcurrentDictionary<string, IModel>();
            _propertiesMap = new ConcurrentDictionary<string, IBasicProperties>();
        }

        /// <summary>
        /// 订阅
        /// </summary>
        /// <param name="name"></param>
        /// <param name="action"></param>
        /// <typeparam name="T"></typeparam>
        /// <returns></returns>
        protected override Task OnSubscribeAsync<T>(string name, Action<T> action)
        {
            var channel = this.CreateChannel(name);
            var consumer = new AsyncEventingBasicConsumer(channel);

            consumer.Received += (model, args) =>
            {
                if (args == default)
                {
                    return Task.CompletedTask;
                }

                var data = this._serializerContext.Deserialize<T>(args.Body);
                action?.Invoke(data);
                //消息确认
                channel.BasicAck(args.DeliveryTag, false);
                return Task.CompletedTask;
            };

            channel.BasicConsume(name, autoAck: false, consumer: consumer);
            return Task.CompletedTask;
        }

        /// <summary>
        /// 发布
        /// </summary>
        /// <param name="name"></param>
        /// <param name="data"></param>
        /// <typeparam name="T"></typeparam>
        /// <returns></returns>
        protected override Task OnPublishAsync<T>(string name, T data)
        {
            var channel = this.CreateChannel(name);
            var properties = this.CreateProperties(channel, name);
            var buffer = this._serializerContext.Serialize(data);

            // 交换机(使用直连默认的)
            // routingKey 内置匹配通道规则(直接为队列名称)
            channel.BasicPublish(string.Empty, name, properties, buffer);
            return Task.CompletedTask;
        }

        /// <summary>
        /// 创建通道消息属性
        /// </summary>
        /// <param name="channel"></param>
        /// <param name="name"></param>
        /// <returns></returns>
        private IBasicProperties CreateProperties(IModel channel, string name)
        {
            if (!this._propertiesMap.TryGetValue(name, out var properties))
            {
                properties = channel.CreateBasicProperties();
                properties.Persistent = true; //持久化消息
                this._propertiesMap.TryAdd(name, properties);
            }

            return properties;
        }

        /// <summary>
        /// 创建信道
        /// </summary>
        /// <param name="name"></param>
        /// <returns></returns>
        private IModel CreateChannel(string name)
        {
            base.Logger.LogDebug($"Creating Channel {name}");

            if (!this._channelMap.TryGetValue(name, out var channel))
            {
                //创建信道
                channel = this._connection.CreateModel();

                //限制向客户端推送消息的流速(每条消息大小，一般设为0，表示不限制,给一个消费者推送多于N个消息知道ack后再推,限制是channel级别的还是consumer级别)
                channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

                //申明队列(队列名称,是否持久化,是否排外,自动删除,参数)
                channel.QueueDeclare(queue: name, durable: true, exclusive: false, autoDelete: false,
                    arguments: default);

                //绑定交换机(队列名称,交换机名称,路由key,参数)
                // channel.QueueBind("", "", "", default);

                this._channelMap.TryAdd(name, channel);
            }

            return channel;
        }

        /// <summary>
        /// 获取链接工厂
        /// </summary>
        /// <returns></returns>
        private ConnectionFactory GetConnectionFactory()
        {
            if (_factory != default)
            {
                return _factory;
            }

            base.Logger.LogDebug($"Creating {nameof(ConnectionFactory)}");

            var factory = new ConnectionFactory()
            {
                HostName = this._configuration.GetValue<string>("RabbitMQ:HostName"),
                UserName = this._configuration.GetValue<string>("RabbitMQ:UserName"),
                Password = this._configuration.GetValue<string>("RabbitMQ:Password"),
                Port = this._configuration.GetValue<int>("RabbitMQ:Port"),
            };

            base.Logger.LogDebug($"Created {nameof(ConnectionFactory)}");

            return factory;
        }
    }
}